package com.lwc.backendsinglestructure.queue.consumer;

import cn.hutool.core.collection.CollectionUtil;
import com.lwc.backendsinglestructure.queue.enums.MessageQueueEnum;
import com.lwc.backendsinglestructure.queue.model.BaseQueueDTO;
import lombok.extern.slf4j.Slf4j;

import java.time.LocalDateTime;
import java.util.Set;
import java.util.concurrent.Executors;
import java.util.concurrent.Semaphore;
import java.util.concurrent.ThreadPoolExecutor;

/**
 * 抽象Redis延迟队列消费者
 */
@Slf4j
public abstract class AbstractRedisDelayQueueConsumer<T extends BaseQueueDTO>  {

    private final boolean isRun = true;

    private boolean isInit = false;

    /**
     * 默认支持并发消费数量 1, 子类可重写修改
     * @return
     */
    protected int getPoolSize(){
        return 1;
    }

    /**
     * 线程池
     */
    private ThreadPoolExecutor threadPoolExecutor;

    /**
     * 初始化
     */
    private void init(){
        if (isInit){
            return;
        }
        initThreadPool(getPoolSize());
        isInit = true;
    }

    /**
     * TODO 线程池创建方式可查询有无优化的地方
     * 初始化线程池
     * @param poolSize
     */
    private void initThreadPool(int poolSize){
        if(threadPoolExecutor == null){
            synchronized (this){
                if(threadPoolExecutor == null){
                    threadPoolExecutor = (ThreadPoolExecutor) Executors.newFixedThreadPool(poolSize);
                }
            }
        }
    }

    /**
     * 核心调度方法
     */
    public void schedule(){
        this.init();
        try {
            Semaphore semaphore = new Semaphore(getPoolSize());
            while(isRun){
                // 判断是否有执行资格
                semaphore.acquire();

                // 从队列里面消费数据 (可能有一批数据)
                Set<String> originalData = this.consumeMessage();
                if(CollectionUtil.isEmpty(originalData)){
                    semaphore.release();
                    continue;
                }
                // 线程池异步执行
                threadPoolExecutor.execute(new Runnable() {
                    @Override
                    public void run() {
                        try {
                            doNextConsume(originalData);
                        } finally {
                            semaphore.release();
                        }
                    }
                });

            }
        } catch (Exception outerEx) {
            log.error("启动消息队列失败, 原因 ", outerEx);
        }
    }

    /**
     * 将字符串转化为对象 子类去重写
     * @param originalData
     * @return
     */
    protected abstract T convert(Set<String> originalData);

    /**
     * 下一步实际处理逻辑
     * @return
     */
    protected abstract void doNextConsume(Set<String> originalData);

    protected String getQueueName(){
        return getMessageQueue().getQueue();
    }

    /**
     * 获取队列信息 由子类重写
     * @return
     */
    protected abstract MessageQueueEnum getMessageQueue();

    /**
     * 从Redis拉取数据消费
     * @return
     */
    private Set<String> consumeMessage() {
        long currentTimeStamp = System.currentTimeMillis();
        String queueName = getQueueName();
        /**
         * 在并发消费下有线程安全问题, 下面的1,2步骤应该在lua脚本里面执行保证原子性
         */
        // 1. 从Redis中获取指定范围的数据  -99999 ------- currentTimeStamp 范围内的数据
        // 2. 删除Redis中的数据
        return null;
    }

}
